

package com.hazelcast.replicatedmap.impl.operation;

import com.hazelcast.internal.namespace.NamespaceUtil;
import com.hazelcast.internal.serialization.impl.SerializationUtil;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.replicatedmap.impl.ReplicatedMapService;
import com.hazelcast.replicatedmap.impl.record.ReplicatedRecordStore;
import com.hazelcast.spi.merge.SplitBrainMergePolicy;
import com.hazelcast.spi.merge.SplitBrainMergeTypes.ReplicatedMapMergeTypes;

import java.io.IOException;
import java.util.List;

/**
 * Contains multiple merging entries for split-brain healing with a {@link SplitBrainMergePolicy}.
 *
 * @since 3.10
 */
public class MergeOperation extends AbstractNamedSerializableOperation {

    private String name;
    private List<ReplicatedMapMergeTypes<Object, Object>> mergingEntries;
    private SplitBrainMergePolicy<Object, ReplicatedMapMergeTypes<Object, Object>, Object> mergePolicy;

    private transient boolean hasMergedValues;

    public MergeOperation() {
    }

    MergeOperation(String name, List<ReplicatedMapMergeTypes<Object, Object>> mergingEntries, SplitBrainMergePolicy<Object, ReplicatedMapMergeTypes<Object, Object>, Object> mergePolicy) {
        this.name = name;
        this.mergingEntries = mergingEntries;
        this.mergePolicy = mergePolicy;
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public void run() {
        ReplicatedMapService service = getService();
        ReplicatedRecordStore recordStore = service.getReplicatedRecordStore(name, true, getPartitionId());

        for (ReplicatedMapMergeTypes<Object, Object> mergingEntry : mergingEntries) {
            if (recordStore.merge(mergingEntry, mergePolicy)) {
                hasMergedValues = true;
            }
        }
    }

    @Override
    public Object getResponse() {
        return hasMergedValues;
    }

    @Override
    protected void writeInternal(ObjectDataOutput out) throws IOException {
        super.writeInternal(out);
        out.writeString(name);
        SerializationUtil.writeList(mergingEntries, out);
        out.writeObject(mergePolicy);
    }

    @Override
    protected void readInternal(ObjectDataInput in) throws IOException {
        super.readInternal(in);
        name = in.readString();
        mergingEntries = SerializationUtil.readList(in);
        mergePolicy = NamespaceUtil.callWithNamespace(in::readObject, name, ReplicatedMapService::lookupNamespace);
    }

    @Override
    public int getClassId() {
        return ReplicatedMapDataSerializerHook.MERGE;
    }
}
